package com.chenjj.bigdata.mapreduce.mapper;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;

import static org.apache.hadoop.io.NullWritable.*;

public class Dingchang2DingjieMapper extends Mapper<Object, Text, Text, NullWritable> {
    private Logger logger = LoggerFactory.getLogger(getClass());
    private MultipleOutputs<Text,NullWritable> mos;
    private Text line = new Text();
    private int[] splitNumber = {1,4,4,1};
    private String splitStr = "|";
    private String encode  = "UTF-8";

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        mos.close();
        super.cleanup(context);
    }

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        mos = new MultipleOutputs(context);
        super.setup(context);
    }

    public void map(Object key, Text value, Context context)
            throws IOException, InterruptedException {
        String fileName = ((FileSplit) context.getInputSplit()).getPath().getName();

        logger.info("Dingchang2Dingjie mapper run,fileName:{},key:{},value:{}",fileName,key,value);
        //需要考虑字符编码的问题
        byte[] src = value.getBytes();
        int lineSize = 0;
        for (int i=0; i<splitNumber.length;i++){
            lineSize += splitNumber[i];
        }

        if(lineSize != src.length){
            throw new RuntimeException("文本行长度（"+src.length+"）不等于约定的行长度("+lineSize+")：" + value.toString());
        }


        StringBuilder stringBuilder = new StringBuilder();
        int index = 0;
        for (int i=0; i< splitNumber.length; i++){
            byte[] temp = new byte[splitNumber[i]];
            //将字节截取之后放入temp中
            System.arraycopy(src,index,temp,0,splitNumber[i]);
            String field = new String(temp,encode);
            field += splitStr;
            stringBuilder.append(field);
            temp = null;
            index += splitNumber[i];
        }
        line.set(stringBuilder.toString());

        //context.write(line,NullWritable.get());
        mos.write("namedOutput",line,NullWritable.get());
    }
}
